package derbyclnt.sockets;

import java.util.Map;
import java.util.PriorityQueue;
import java.util.Queue;
import java.util.concurrent.atomic.AtomicBoolean;

import com.google.protobuf.MessageLite;

import derbyclnt.MainManager;
import derbyclnt.messages.HelloProtos;
import derbyclnt.messages.MessageTypesProtos;
import derbyclnt.messages.MessageTypesProtos.MessageTypes;
import static derbyclnt.Logger.*;

public class SocketManager {

	private Map<MessageTypes, MessageProcessor> _messageProcessors = MessageProcessor.registerMessageProcessors();
	private AsyncClientSocket _socket;
	AtomicBoolean _socketValid = new AtomicBoolean(false);
	private Queue<RawData> _writingRaws;
	private Queue<RawData> _readingRaws;

	private AtomicBoolean _readingStarted = new AtomicBoolean(false);
	private RawData _readingData = null;
	private AtomicBoolean _writingStarted = new AtomicBoolean(false);
	private RawData _writingData = null;
	
	private Object _writingMutex = new Object();
	private Object _readingMutex = new Object();

	private enum ProtocolModes {
		MODE_META, MODE_DATA
	}

	private ProtocolModes _writingMode, _readingMode;

	public SocketManager() {
		_writingRaws = new PriorityQueue<RawData>();
		_readingRaws = new PriorityQueue<RawData>();
	}

	public void start() throws Exception {
		logd("SocketManager::start");
		_writingMode = ProtocolModes.MODE_META;
		_readingMode = ProtocolModes.MODE_META;
		
		assert (_socketValid.get() == false);

		logi("Connecting to server... ");
		_socket = new AsyncClientSocket("192.168.1.100", 7788,
				new SocketManager.ClientSocketListener()); // TODO
		_socket.connect();
		logd("SocketManager#start() finished");
	}

	public void stop() {
		_socketValid.set(false);
		_socket.disconnect();
		assert (!_socket.isValid());
	}

	// TODO synchronization!!!
	public void send(final MessageLite message, MessageTypesProtos.MessageTypes type) throws Exception {
		if (!_socketValid.get())
			throw new Exception("Socket is invalid. ");
		assert (_socket.isValid());
		synchronized (_writingMutex) {
			_writingRaws.add(new RawData(message, type));
		}
		if (!this._writingStarted.get()) {
			startWriting();
		}
	}

	public void receiveAll(MainManager mgr) throws Exception {
		if (!_socketValid.get())
			throw new Exception("Socket is invalid. ");
		assert (_socket.isValid());
		synchronized (_readingMutex) {
			for (RawData raw : _readingRaws) {
				// convert to message
				_messageProcessors.get(raw.getType()).Process(mgr, raw);
			}
			_readingRaws.clear();
		}
		
		//if (this._readingStarted.get() == false) {
		//	startReading();
		//}
	}

	
	private void startReading() {
		this._readingStarted.set(true);
		try {
			if (_readingMode == ProtocolModes.MODE_META) {
				assert _readingData == null;
				_socket.receive(RawData.META_BYTE_SIZE);
			} else if (_readingMode == ProtocolModes.MODE_DATA) {
				assert _readingData != null;
				_socket.receive(_readingData.getSize());
			} else {
				assert false : "mode is illegal!";
			}
		} catch (Exception e) {
			assert (false);
			loge(e.getMessage());
			this._readingStarted.set(false);
		}
	}
	
	private void startWriting () {
		this._writingStarted.set(true);
		try {
			if (_writingMode == ProtocolModes.MODE_META) {
				assert _writingData == null;
				synchronized (_writingMutex) {
					if (!_writingRaws.isEmpty()) {
						_writingData = _writingRaws.poll();
						_socket.send(_writingData.getMetaData());
					} else {
						this._writingStarted.set(false);
					}
				}
			} else if (_writingMode == ProtocolModes.MODE_DATA) {
				assert _writingData != null;
				_socket.send(_writingData.getData());
			} else {
				assert false;
			}
		} catch (Exception e) {
			assert false;
			loge(e.getMessage());
		}
	}

	private class ClientSocketListener implements SocketListener {

		@Override
		public void onDataReceived(byte[] data, boolean success) {
			logd("onDataReceived#success = " + success);
			if (!success) {
				_socketValid.set(false);
				return;
			}
			
			try {
				if (_readingMode == ProtocolModes.MODE_META) {
					assert _readingData == null;
					_readingData = new RawData(data);
					_readingMode = ProtocolModes.MODE_DATA;
					startReading();
				} else if (_readingMode == ProtocolModes.MODE_DATA) {
					assert _readingData != null;
					assert(data.length == _readingData.getSize());
					_readingData.setData(data);
					synchronized (_readingMutex) {
						_readingRaws.add(_readingData);
						_readingData = null;
					}
					_readingMode = ProtocolModes.MODE_META;
					startReading();
				} else {
					assert false;
				}
			} catch (Exception e) {
				loge(e.getMessage());
			}			
		}

		@Override
		public void onDataSent(boolean success) {
			logd("onDataSend#success=" + success);
			if (!success) {
				_socketValid.set(false);
				return;
			}
			try {
				if (_writingMode == ProtocolModes.MODE_META) {
					assert _writingData != null;
					_writingMode = ProtocolModes.MODE_DATA;
					startWriting();
				} else if (_writingMode == ProtocolModes.MODE_DATA) {
					assert _writingData != null;
					_writingData = null;
					_writingMode = ProtocolModes.MODE_META;
					startWriting();
				} else {
					assert false;
				}
				
			} catch (Exception e) {
				loge(e.getMessage());
			} 
		}

		@Override
		public void onClosedStatus(boolean isClosed) {
			logd("onClosedStatus#isClosed = " + isClosed);
			_socketValid.set(!isClosed);
			if (isClosed)
				return;
			
			assert _readingStarted.get() == false;
			startReading();

			// TODO
			HelloProtos.Hello.Builder helloBuilder = HelloProtos.Hello.newBuilder();
			helloBuilder.setClientId(-1);
			helloBuilder.setText("Hi, server!! ");
			try {
				send(helloBuilder.build(), MessageTypes.MESSAGE_HELLO);
			} catch (Exception e) {
				loge(e.getMessage());
			}
		}
	}

}
